package com.example.kafka.consumer;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

/**
 * @application:
 * @author: LiYuhang
 * @Date: 2021/6/26 14:45
 * @version: 0.1
 * @Description :
 * @Modified By:
 */
@Slf4j
public class ConsumerDemo3 implements Consumer {

    private final KafkaConsumer<String, String> consumer;
    private final String topic;


    public ConsumerDemo3(String topic,String groupID) {
        Properties defaultConfig = Consumer.getDefaultConfig(groupID);
        defaultConfig.put("max.poll.records",1);
        this.consumer = new KafkaConsumer<String, String>(defaultConfig);
        this.topic = topic;
    }


    public static void main(String[] args)  throws Exception{

        ConsumerDemo3 demo1 = new ConsumerDemo3("test","topic.group1");

        readMessageByOffset(demo1.consumer,demo1.topic,2,12);

       //showInfo(demo1.consumer);

    }

    /**
     * 不改变offset方式读取消息
     * @param consumer
     * @param topic
     * @param partition
     * @param offset
     */
    private static void readMessageByOffset(KafkaConsumer<String, String> consumer,String topic,
                                            int partition,int offset) {
        final TopicPartition topicPartition = new TopicPartition(topic,partition);
        consumer.assign(Arrays.asList(topicPartition));
        //不改变当前offset
        consumer.seek(topicPartition, offset);
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(5000));
        System.out.println("##################### isEmpty="+records.isEmpty()+" count:"+records.count());
        for (ConsumerRecord<String, String> record : records) {
            System.out.println(record.toString());
        }
    }


    private static void showInfo( KafkaConsumer<String, String> consumer) {
        System.out.println("#############################");
        System.out.println(consumer.listTopics());
        System.out.println("#############################");
        System.out.println(consumer.partitionsFor("test"));
        System.out.println("#############################");
        System.out.println(consumer.metrics());
    }


}
